<?php

namespace app\mqtt\controller;

use app\common\controller\Base;
use think\facade\Request;
use think\Db;
use think\db\Where;

/**
 * @title 消息服务接口
 * @description 消息服务说明
 * @group mqtt
 */
class MsgServer extends Base
{
    protected $sys_chat_id = 0;

    /**
     * @title 登录
     * @description 登录
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/login
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:account_id type:Int require:1 default:0 other: desc:帐号id
     * @param name:account_type type:Int require:1 default:0 other: desc:帐号类型
     * @param name:mqtt_client_id type:Int require:1 default:0 other: desc:mqtt_client_id
     *
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function login()
    {
        $model = new \app\mqtt\model\MqttClientOnline();
        $Fields = $model->getTableFields();
        $data = \think\facade\Request::only($Fields);
        $data['last_time'] = time();
        $data['online'] = 1;
        if (!isset($data['token']) || !isset($data['account_id']) || !isset($data['mqtt_client_id'])) {
            $msg['msg'] = '验证失败';
            $msg['validate'] = "缺少参数";
            return $this->errorJson($msg);
        }
        $online = Db::name('mqtt_client_online')->where("account_id", $data['account_id'])->where('account_type', $data['account_type'])->find();
        if (!$online) {

            $insert_res = Db::name('mqtt_client_online')->data($data)->insert();
            if (!$insert_res) {

                //TODO 创建系统消息，登录失败
                $msg['msg'] = '登录失败,插入数据失败！';
                return $this->errorJson($msg);
            }
        } else if (!$online['online']) {
            $update_map = $data;
            unset($update_map['account_id']);
            $update_res = Db::name('msg')->where('account_id', $data['account_id'])->where('account_type', $data['account_type'])->update($update_map);

            if (!$update_res) {

                //TODO 创建系统消息，登录失败

                $msg['msg'] = '登录失败,更新据失败！';
                return $this->errorJson($msg);
            }
        }

        //TODO 创建系 统消息，登录成功


        $msg['data'] = $data;
        //接口输出json数据
        $msg['msg'] = '登录成功,操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 创建窗口
     * @description 创建窗口
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/create_chat
     * @method post
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:title type:String require:1 default:1 other: desc:标题
     * @param name:create_account_id type:Int require:1 default:0 other: desc:创建者
     * @param name:create_account_type type:String require:1 default:0 other: desc:创建者类型
     *
     * @param name:main_account_id type:Int require:1 default:0 other: desc:管理员
     * @param name:main_account_type type:String require:1 default:0 other: desc:管理员类型
     * @param name:logo type:String require:0 default:0 other: desc:logo
     * @param name:info type:String require:0 default:0 other: desc:简介
     * @param name:chat_type type:Int require:1 default:0 other: desc:窗口类型（0-客服窗口；1-单聊窗口；2-群聊窗口）
     * @param name:member_account_ids type:String require:0 default:0 other:  desc:类型_id，参与用户帐号（逗号隔开）
     *
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function create_chat()
    {
        $data = input();
        $validate = new \think\Validate;
        $validate->rule([
            'title' => 'require',
            'main_account_id' => 'require|number',
//                'chat_type' => 'require'
        ]);
        if (!$validate->check($data)) {
            $msg['msg'] = $validate->getError();
            return $this->errorJson($msg);
        }
        if (!isset($data['create_account_id'])) {
            $data['create_account_id'] = $data['main_account_id'];
            $data['create_account_type'] = $data['main_account_type'];
        }
        //创建窗口
        $chat_id = Db::name("chat")->strict(false)->insertGetId($data);

        //添加聊天用户
        $members = [];
        //测试main_account_id为150036
        $member_account = get_user($data['main_account_id'], $data['main_account_type']);
        if ($member_account) {
            $members[] = $member_account;
        }
        $member_account_ids = explode(",", $data['member_account_ids']);

        foreach ($member_account_ids as $member_account_id) {
            $member_account_id = explode("_", $member_account_id);
            if($data['main_account_id']==$member_account_id[1]&&$data['main_account_type']==$member_account_id[0]){//去重
                continue;
            }
            $member_account = get_user($member_account_id[1], $member_account_id[0]);
            if ($member_account) {
                $members[] = $member_account;
            }
        }


        $chat_members = [];

        foreach ($members as $member) {
            $chat_member = (array)$member;
            $chat_member['add_time'] = time();
            $chat_member['in_accept'] = true;
            $chat_member['chat_type'] = $data['chat_type'];
            $chat_member['chat_id'] = $chat_id;
            $chat_members[] = $chat_member;
        }

        Db::name("chat_member")->strict(false)->insertAll($chat_members);
        $chat = Db::name("chat")->where('id', $chat_id)->find();
        $chat_members = Db::name("chat_member")->where('chat_id', $chat_id)->select();

        $chat['members'] = $chat_members;
        $msg['data'] = $chat;

        //接口输出json数据
        $msg['msg'] = '操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 获取用户所有窗口
     * @description 获取用户所有窗口
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/account_chats
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:account_id type:Int require:1 default:0 other: desc:帐号id
     * @param name:account_type type:Int require:1 default:0 other: desc:帐号类型
     * @param name:status type:Int require:0 default:0 other: desc:状态（0-禁用；1-启用；）
     * @param name:is_deleted type:Int require:0 default:0 other: desc:已删除（0-否；1-是；）
     * @param name:title type:String require:0 default:0 other: desc:标题
     * @param name:can_send type:Int require:0 default:0 other: desc:可发送（0-否；1-是；）
     * @param name:chat_type type:Int require:0 default:0 other: desc:窗口类型（0-客服窗口；1-单聊窗口；2-群聊窗口）
     * @param name:in_accept type:Int require:0 default:0 other: desc:可接收（0-否；1-是；）
     *
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function account_chats()
    {
        $model = new \app\mqtt\model\ChatMember();
        $Fields = $model->getTableFields();
        $data = \think\facade\Request::only($Fields);
        if (!isset($data['account_id'])) {
            $msg['msg'] = '验证失败';
            $msg['validate'] = "缺少参数";
            return $this->errorJson($msg);
        }
        $where_map = $data;
        $chat_member = Db::name('chat_member')->where($where_map)->select();
        $msg['data'] = $chat_member;
        //接口输出json数据
        $msg['msg'] = '操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 获取窗口的用户消息列表
     * @description 获取窗口的用户消息列表
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/account_msgs
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:send_account_id type:Int require:0 default:0 other: desc:发送帐号id
     * @param name:accept_account_id type:Int require:0 default:0 other: desc:接收用户
     * @param name:send_account_type type:String require:0 default:0 other: desc:发送帐号类型
     * @param name:accept_account_type type:String require:0 default:0 other: desc:接收用户类型
     * @param name:is_sended type:Int require:0 default:0 other: desc:已发送（0-否；1-是；）
     * @param name:content type:String require:0 default:0 other: desc:内容
     * @param name:chat_id type:int require:0 default:0 other: desc:窗口id
     * @param name:is_file type:Int require:0 default:0 other: desc:文件（0-否；1-是；）
     * @param name:file_name type:String require:0 default:0 other: desc:文件名
     * @param name:file_type type:String require:0 default:0 other: desc:文件类型
     * @param name:is_accepted type:Int require:0 default:0 other: desc:已接受（0-否；1-是；）
     * @param name:msg_type type:String require:0 default:0 other: desc:消息类型
     * @param name:msg_id type:String require:0 default:0 other: desc:msg_id
     *
     *
     * @param name:page type:Int require:0 default:0 other: desc:当前页
     * @param name:pagesize type:Int require:0 default:0 other: desc:每页数量
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function account_msgs()
    {

        $model = new \app\mqtt\model\AccountMsg();
        $Fields = $model->getTableFields();
        $data = \think\facade\Request::only($Fields);

        $where_map = $data;

        unset($where_map['token']);
        foreach ($where_map as $k => $v){
            if(empty($v)){
                unset($where_map[$k]);
            }
        }

        $update_map = [
            ['chat_id','=',$data['chat_id']],
            ['accept_account_id','=',$data['accept_account_id']],
            ['accept_account_type','=',$data['accept_account_type']],
        ];

//        $un_accepted_num = Db::name("account_msg")
//            ->where($update_map)
//            ->where('is_accepted',0)
//            ->where('send_account_id','<>',$data['accept_account_id'])
//            ->count();

//        halt($un_accepted_num);

//        if($un_accepted_num < 15){
//            $chat_member['data'] = Db::name('account_msg')
//                ->where($where_map)
//                ->order('create_time','desc')
//                ->limit($un_accepted_num)
//                ->select();
//        }else{
//            $chat_member = Db::name('account_msg')
//                ->where($where_map)
//                ->order('create_time','desc')
//                ->paginate();
//        }

        $chat_member = Db::name('account_msg')
                ->where($where_map)
                ->order('create_time','desc')
                ->paginate();

//        halt($chat_member);

        //这个接口 /mqtt/msg_server/account_msgs
        //对方发送的消息确认已读
        $accept_time = time();
        Db::name("account_msg")
            ->where($update_map)
            ->where('send_account_id','<>',$data['accept_account_id'])
            ->update(['is_accepted' => 1,'accept_time'=>$accept_time,'is_sended'=>1]);


        $msg['data'] = $chat_member;
        //接口输出json数据
        $msg['msg'] = '操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 统计用户未读消息数目
     * @description 统计用户未读消息数目
     * @author 开发者（袁丽蓉）
     * @url /mqtt/msg_server/get_unread_num
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
//     * @param name:in_service type:TinyInt require:1 default:0 other: desc:服务中
//     * @param name:is_delete type:TinyInt require:1 default:0 other: desc:已删除
     * @param name:chat_id type:Int require:0 default:0 other: desc:窗口id
     * @param name:accept_account_id type:Varchar require:1 default:0 other: desc:接收者id
     * @param name:accept_account_type type:Varchar require:1 default:0 other: desc:接收者类型

     *
//     * @return in_service:服务中
//     * @return is_delete:已删除
     * @return chat_id:窗口id
     * @return accept_account_id:接收者id
     * @return accept_account_type:接收者类型
     *
     */
    public function get_unread_num (){
        //模块 - 输入数据
        $input = input();

        // ---	模块 - 查询  ---
        //查询条件
        $temp_map = [
            ['chat_id','=',$input['chat_id']],
            ['accept_account_id','=',$input['accept_account_id']],
            ['accept_account_type','=',$input['accept_account_type']]
        ];

        //去除为空的条件
        $select_map = [];
        foreach ($temp_map as $k=>$v){
            if ($v[2]===null || $v[2]=='' || $v[2]=='%%'){
                continue;
            }
            array_push($select_map,$v);

        }

        //过滤系统消息
        $find_map = new Where();
        $find_map['msg_type'] = ['not in',['create_chat','in_service']];

        //获取用户未读消息数量
        $unread_num = Db::name('account_msg')
                    ->where($find_map)
                    ->where($select_map)
                    ->where('send_account_id','<>',$input['accept_account_id'])
                    ->where('is_accepted', 0)
                    ->count();

        //获取用户未读消息列表
        $list = Db::name('account_msg')
            ->where($find_map)
            ->where($select_map)
            ->where('send_account_id','<>',$input['accept_account_id'])
            ->where('is_accepted', 0)
            ->select();

        // count 返回数据
        $msg['num'] = $unread_num;
        $msg['list'] = $list;

        //接口输出json数据
        $msg['msg'] = '操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 添加聊天用户
     * @description 添加聊天用户
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/add_member
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:chat_id type:String require:1 default:0 other: desc:窗口id
     * @param name:account_id type:Int require:1 default:0 other: desc:帐号id
     * @param name:account_type type:String require:1 default:0 other: desc:帐号类型
     * @param name:status type:Int require:0 default:0 other: desc:状态（0-禁用；1-启用；）
     * @param name:is_deleted Int:String require:0 default:0 other: desc:已删除（0-否；1-是；）
     * @param name:can_send type:int require:0 default:0 other: desc:可发送（0-否；1-是；）
     * @param name:in_accept type:Int require:0 default:0 other: desc:可接收（0-否；1-是；）
     * @param name:info type:String require:0 default:0 other: desc:简介
     * @param name:chat_type type:String require:1 default:0 other: desc:窗口类型
     * @param name:nickname type:String require:0 default:0 other: desc:昵称
     *
     *
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function add_member()
    {

        $model = new \app\mqtt\model\ChatMember();
        $Fields = $model->getTableFields();
        $data = \think\facade\Request::only($Fields);

        $validate = new \think\Validate;
        $validate->rule([
            'chat_id' => 'require|number',
            'account_id' => 'require|number',
            'chat_type' => 'require'
        ]);
        if (!$validate->check($data)) {
            $msg['msg'] = $validate->getError();
            return $this->errorJson($msg);
        }
        $chat_member_id = Db::name('chat_member')->insertGetId($data);
        $chat_member = Db::name('chat_member')->find($chat_member_id);
        $msg['data'] = $chat_member;
        //接口输出json数据
        $msg['msg'] = '操作成功！';
        return $this->successJson($msg);
    }

    /**
     * @title 发送窗口消息
     * @description 发送窗口消息
     * @author 开发者（梁敏）
     * @url /mqtt/msg_server/send_msg
     * @method *
     *
     * @param name:token require:1 default: desc:token
     *
     * @param name:chat_id type:Int require:1 default:1 other: desc:窗口id
     * @param name:send_account_id type:Int require:1 default:1 other: desc:帐号id
     * @param name:send_account_type type:String require:1 default:1 other: desc:帐号类型
     * @param name:content type:String require:1 default:0 other: desc:内容
     * @param name:msg_type type:String require:1 default:0 other: desc:普通文件，语音，视频，图片，链接
     * @param name:accept_account_id type:String require:0 default:0 other: desc:接收用户，类型_id，逗号隔开
     * @param name:is_file type:int require:0 default:0 other: desc:是文件（0-否；1-是；）
     * @param name:file_name type:String require:0 default:0 other: desc:文件名
     * @param name:file_size type:String require:0 default:0 other: desc:文件大小
     * @param name:file_type type:String require:1 default:0 other: desc:文件类型（普通文件，语音，视频，图片）
     * @param name:file_url type:String require:0 default:0 other: desc:文件路径
     *
     *
     * @return status:状态
     * @return msg:msg
     * @return data:data
     */
    public function send_msg()
    {

        $model = new \app\mqtt\model\Msg();
        $Fields = $model->getTableFields();
        $data = \think\facade\Request::only($Fields);
//        logs("22222222222222222222",$data);
        $validate = new \think\Validate;
        $validate->rule([
            'chat_id' => 'integer',
            'send_account_id' => 'number',
            'content' => 'require',
            'msg_type' => 'require',
            'accept_account_id' => 'require'
        ]);
        if (!$validate->check($data)) {
            $msg['msg'] = $validate->getError();
            return $this->errorJson($msg);
        }

        //添加msg
        $data['create_time'] = time();

        $msg_id = Db::name('msg')->insertGetId($data);
        $data = Db::name('msg')->find($msg_id);

        $accept_account_ids = explode(",", $data['accept_account_id']);
        $accept_account_ids = array_unique($accept_account_ids);
//        logs("accept_account_ids", $accept_account_ids);
        //添加account_msgs
        $account_msgs = [];
        foreach ($accept_account_ids as $accept_account_id) {
            $accept_account_id = explode("_", $accept_account_id);
            $account_msg = $data;
            unset($account_msg['id']);
            $account_msg['accept_account_id'] = $accept_account_id[1];
            $account_msg['accept_account_type'] = $accept_account_id[0];
            $account_msg['send_account_type'] = $data['send_account_type'];
            $account_msg['create_time'] = time();
            $account_msg['msg_id'] = $msg_id;
            $account_msgs[] = $account_msg;
        }

        Db::name("account_msg")->insertAll($account_msgs);

        //这个接口 /mqtt/msg_server/send_msg
        //自己发送自己接收的消息，发送就改变对应的状态
        Db::name("account_msg")
            ->where('msg_id',$msg_id)
            ->where('accept_account_id',$data['send_account_id'])
            ->update(['is_accepted' => 1,'accept_time'=>time(),'is_sended'=>1]);

        $account_msgs = Db::name("account_msg")->where('msg_id', $msg_id)->select();

        //系统消息中订单列表

//        $table=config('database.zb_service');
//
//
//
//        $order_lists=Db::connect($table)->name('service_order')->where('chat_id',$data['chat_id'])->select();
////        dump(Db::connect($table)->name('service_order')->getLastSql());
//
//        foreach ($order_lists as $key=>$order){
//            $order_lists[$key]['order_info']=json_decode($order['order_info'],true);
//            $order=$order_lists[$key]['order_info']['order'];
//            $order_item=$order_lists[$key]['order_info']['order_item'];
//        }
//
//        $msg['order']=$order;
//        //订单、商品详情详情
//        $msg['order_item']=$order_item;
//        $msg['pust_time']=time();          //订单推送的时间
////        $msg['title']='新的订单';           //标题

        $msg['data'] = $data;
        $msg['data']['account_msgs'] = $account_msgs;
        //接口输出json数据
        $msg['msg'] = '发送失败！';

        foreach ($account_msgs as $account_msg) {
            $mqtt = new Mqtt('47.99.158.37', 1883, 'clientx9293670xxctr');
            if ($mqtt->connect()) {
                $topic = "msg/" . $account_msg['chat_id'] . "/" . $account_msg['accept_account_type'] . "_" . $account_msg['accept_account_id'];

                $mqtt->publish($topic, json_encode($account_msg), 2);
                logs("topic", $topic);
                logs("topic", $account_msg);
                $msg['msg'] = '发送成功！';

//            $mqtt->publish("xxx3809293670ctr", "setr=3xxxxxxxxx", 0);
                // 发送到 xxx3809293670ctr 的主题 一个信息 内容为 setr=3xxxxxxxxx Qos 为 0
                $mqtt->close();    //发送后关闭链接
            } else {
                echo "Time out!\n";
            }
        }

        return $this->successJson($msg);
    }

}

/* phpMQTT */

class Mqtt
{

    private $socket;             /* holds the socket    */
    private $msgid = 1;            /* counter for message id */
    public $keepalive = 10;        /* default keepalive timmer */
    public $timesinceping;        /* host unix time, used to detect disconects */
    public $topics = array();     /* used to store currently subscribed topics */
    public $debug = false;        /* should output debug messages */
    public $address;            /* broker address */
    public $port;                /* broker port */
    public $clientid;            /* client id sent to brocker */
    public $will;                /* stores the will of the client */
    private $username;            /* stores username */
    private $password;            /* stores password */

    public $cafile;

    function __construct($address, $port, $clientid, $cafile = NULL)
    {
        $this->broker($address, $port, $clientid, $cafile);
    }

    /* sets the broker details */
    function broker($address, $port, $clientid, $cafile = NULL)
    {
        $this->address = $address;
        $this->port = $port;
        $this->clientid = $clientid;
        $this->cafile = $cafile;
    }

    function connect_auto($clean = true, $will = NULL, $username = NULL, $password = NULL)
    {
        while ($this->connect($clean, $will, $username, $password) == false) {
            sleep(10);
        }
        return true;
    }

    /* connects to the broker
        inputs: $clean: should the client send a clean session flag */
    function connect($clean = true, $will = NULL, $username = NULL, $password = NULL)
    {

        if ($will) $this->will = $will;
        if ($username) $this->username = $username;
        if ($password) $this->password = $password;


        if ($this->cafile) {
            $socketContext = stream_context_create(["ssl" => [
                "verify_peer_name" => true,
                "cafile" => $this->cafile
            ]]);
            $this->socket = stream_socket_client("tls://" . $this->address . ":" . $this->port, $errno, $errstr, 60, STREAM_CLIENT_CONNECT, $socketContext);
        } else {
            $this->socket = stream_socket_client("tcp://" . $this->address . ":" . $this->port, $errno, $errstr, 60, STREAM_CLIENT_CONNECT);
        }

        if (!$this->socket) {
            if ($this->debug) error_log("stream_socket_create() $errno, $errstr \n");
            return false;
        }

        stream_set_timeout($this->socket, 5);
        stream_set_blocking($this->socket, 0);

        $i = 0;
        $buffer = "";

        $buffer .= chr(0x00);
        $i++;
        $buffer .= chr(0x06);
        $i++;
        $buffer .= chr(0x4d);
        $i++;
        $buffer .= chr(0x51);
        $i++;
        $buffer .= chr(0x49);
        $i++;
        $buffer .= chr(0x73);
        $i++;
        $buffer .= chr(0x64);
        $i++;
        $buffer .= chr(0x70);
        $i++;
        $buffer .= chr(0x03);
        $i++;

        //No Will
        $var = 0;
        if ($clean) $var += 2;

        //Add will info to header
        if ($this->will != NULL) {
            $var += 4; // Set will flag
            $var += ($this->will['qos'] << 3); //Set will qos
            if ($this->will['retain']) $var += 32; //Set will retain
        }

        if ($this->username != NULL) $var += 128;    //Add username to header
        if ($this->password != NULL) $var += 64;    //Add password to header

        $buffer .= chr($var);
        $i++;

        //Keep alive
        $buffer .= chr($this->keepalive >> 8);
        $i++;
        $buffer .= chr($this->keepalive & 0xff);
        $i++;

        $buffer .= $this->strwritestring($this->clientid, $i);

        //Adding will to payload
        if ($this->will != NULL) {
            $buffer .= $this->strwritestring($this->will['topic'], $i);
            $buffer .= $this->strwritestring($this->will['content'], $i);
        }

        if ($this->username) $buffer .= $this->strwritestring($this->username, $i);
        if ($this->password) $buffer .= $this->strwritestring($this->password, $i);

        $head = "  ";
        $head{0} = chr(0x10);
        $head{1} = chr($i);

        fwrite($this->socket, $head, 2);
        fwrite($this->socket, $buffer);

        $string = $this->read(4);

        if (ord($string{0}) >> 4 == 2 && $string{3} == chr(0)) {
            if ($this->debug) echo "Connected to Broker\n";
        } else {
            error_log(sprintf("Connection failed! (Error: 0x%02x 0x%02x)\n",
                ord($string{0}), ord($string{3})));
            return false;
        }

        $this->timesinceping = time();

        return true;
    }

    /* read: reads in so many bytes */
    function read($int = 8192, $nb = false)
    {

        //    print_r(socket_get_status($this->socket));

        $string = "";
        $togo = $int;

        if ($nb) {
            return fread($this->socket, $togo);
        }

        while (!feof($this->socket) && $togo > 0) {
            $fread = fread($this->socket, $togo);
            $string .= $fread;
            $togo = $int - strlen($string);
        }


        return $string;
    }

    /* subscribe: subscribes to topics */
    function subscribe($topics, $qos = 0)
    {
        $i = 0;
        $buffer = "";
        $id = $this->msgid;
        $buffer .= chr($id >> 8);
        $i++;
        $buffer .= chr($id % 256);
        $i++;

        foreach ($topics as $key => $topic) {
            $buffer .= $this->strwritestring($key, $i);
            $buffer .= chr($topic["qos"]);
            $i++;
            $this->topics[$key] = $topic;
        }

        $cmd = 0x80;
        //$qos
        $cmd += ($qos << 1);


        $head = chr($cmd);
        $head .= chr($i);

        fwrite($this->socket, $head, 2);
        fwrite($this->socket, $buffer, $i);
        $string = $this->read(2);

        $bytes = ord(substr($string, 1, 1));
        $string = $this->read($bytes);
    }

    /* ping: sends a keep alive ping */
    function ping()
    {
        $head = " ";
        $head = chr(0xc0);
        $head .= chr(0x00);
        fwrite($this->socket, $head, 2);
        if ($this->debug) echo "ping sent\n";
    }

    /* disconnect: sends a proper disconect cmd */
    function disconnect()
    {
        $head = " ";
        $head{0} = chr(0xe0);
        $head{1} = chr(0x00);
        fwrite($this->socket, $head, 2);
    }

    /* close: sends a proper disconect, then closes the socket */
    function close()
    {
        $this->disconnect();
        stream_socket_shutdown($this->socket, STREAM_SHUT_WR);
    }

    /* publish: publishes $content on a $topic */
    function publish($topic, $content, $qos = 0, $retain = 0)
    {

        $i = 0;
        $buffer = "";

        $buffer .= $this->strwritestring($topic, $i);

        //$buffer .= $this->strwritestring($content,$i);

        if ($qos) {
            $id = $this->msgid++;
            $buffer .= chr($id >> 8);
            $i++;
            $buffer .= chr($id % 256);
            $i++;
        }

        $buffer .= $content;
        $i += strlen($content);


        $head = " ";
        $cmd = 0x30;
        if ($qos) $cmd += $qos << 1;
        if ($retain) $cmd += 1;

        $head{0} = chr($cmd);
        $head .= $this->setmsglength($i);

        fwrite($this->socket, $head, strlen($head));
        fwrite($this->socket, $buffer, $i);

    }

    /* message: processes a recieved topic */
    function message($msg)
    {
        $tlen = (ord($msg{0}) << 8) + ord($msg{1});
        $topic = substr($msg, 2, $tlen);
        $msg = substr($msg, ($tlen + 2));
        $found = 0;
        foreach ($this->topics as $key => $top) {
            if (preg_match("/^" . str_replace("#", ".*",
                    str_replace("+", "[^\/]*",
                        str_replace("/", "\/",
                            str_replace("$", '\$',
                                $key)))) . "$/", $topic)) {
                if (is_callable($top['function'])) {
                    call_user_func($top['function'], $topic, $msg);
                    $found = 1;
                }
            }
        }

        if ($this->debug && !$found) echo "msg recieved but no match in subscriptions\n";
    }

    /* proc: the processing loop for an "allways on" client
        set true when you are doing other stuff in the loop good for watching something else at the same time */
    function proc($loop = true)
    {

        if (1) {
            $sockets = array($this->socket);
            $w = $e = NULL;
            $cmd = 0;

            //$byte = fgetc($this->socket);
            if (feof($this->socket)) {
                if ($this->debug) echo "eof receive going to reconnect for good measure\n";
                fclose($this->socket);
                $this->connect_auto(false);
                if (count($this->topics))
                    $this->subscribe($this->topics);
            }

            $byte = $this->read(1, true);

            if (!strlen($byte)) {
                if ($loop) {
                    usleep(100000);
                }

            } else {

                $cmd = (int)(ord($byte) / 16);
                if ($this->debug) echo "Recevid: $cmd\n";

                $multiplier = 1;
                $value = 0;
                do {
                    $digit = ord($this->read(1));
                    $value += ($digit & 127) * $multiplier;
                    $multiplier *= 128;
                } while (($digit & 128) != 0);

                if ($this->debug) echo "Fetching: $value\n";

                if ($value)
                    $string = $this->read($value);

                if ($cmd) {
                    switch ($cmd) {
                        case 3:
                            $this->message($string);
                            break;
                    }

                    $this->timesinceping = time();
                }
            }

            if ($this->timesinceping < (time() - $this->keepalive)) {
                if ($this->debug) echo "not found something so ping\n";
                $this->ping();
            }


            if ($this->timesinceping < (time() - ($this->keepalive * 2))) {
                if ($this->debug) echo "not seen a package in a while, disconnecting\n";
                fclose($this->socket);
                $this->connect_auto(false);
                if (count($this->topics))
                    $this->subscribe($this->topics);
            }

        }
        return 1;
    }

    /* getmsglength: */
    function getmsglength(&$msg, &$i)
    {

        $multiplier = 1;
        $value = 0;
        do {
            $digit = ord($msg{$i});
            $value += ($digit & 127) * $multiplier;
            $multiplier *= 128;
            $i++;
        } while (($digit & 128) != 0);

        return $value;
    }


    /* setmsglength: */
    function setmsglength($len)
    {
        $string = "";
        do {
            $digit = $len % 128;
            $len = $len >> 7;
            // if there are more digits to encode, set the top bit of this digit
            if ($len > 0)
                $digit = ($digit | 0x80);
            $string .= chr($digit);
        } while ($len > 0);
        return $string;
    }

    /* strwritestring: writes a string to a buffer */
    function strwritestring($str, &$i)
    {
        $ret = " ";
        $len = strlen($str);
        $msb = $len >> 8;
        $lsb = $len % 256;
        $ret = chr($msb);
        $ret .= chr($lsb);
        $ret .= $str;
        $i += ($len + 2);
        return $ret;
    }

    function printstr($string)
    {
        $strlen = strlen($string);
        for ($j = 0; $j < $strlen; $j++) {
            $num = ord($string{$j});
            if ($num > 31)
                $chr = $string{$j}; else $chr = " ";
            printf("%4d: %08b : 0x%02x : %s \n", $j, $num, $num, $chr);
        }
    }
}